package com.atguigu.tingshu.album.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.RandomUtil;
import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.mapper.TrackInfoMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.cache.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.model.album.TrackInfo;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.*;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.CacheEvict;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"all"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private KafkaService kafkaService;

    /**
     * 保存专辑
     * 1.将提交专辑信息封装专辑对象AlbumInfo对象，向专辑表中增加一条记录
     * 2.将提交专辑标签封装专辑标签集合对象，向专辑标签关系表中增加若干条记录
     * 3.固定为专辑保存四条统计信息
     *
     * @param userId      用户ID
     * @param albumInfoVo 专辑信息
     */
    @Override
    @Transactional(rollbackFor = Exception.class) //默认事务注解当发生RuntimeException或者ERROR会进行事务回滚
    public void saveAlbumInfo(Long userId, AlbumInfoVo albumInfoVo) {
        //1.将提交专辑信息封装专辑对象AlbumInfo对象，向专辑表中增加一条记录
        //1.1 将提交专辑VO转为专辑PO对象
        AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);
        //1.2 为专辑中属性赋值 用户ID，包含声音数量、免费试听集数、状态等
        albumInfo.setUserId(userId);
        albumInfo.setIncludeTrackCount(0);
        albumInfo.setTracksForFree(5);
        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
        //1.3 保存专辑 得到专辑ID
        albumInfoMapper.insert(albumInfo);
        Long albumId = albumInfo.getId();

        //2.将提交专辑标签封装专辑标签集合对象，向专辑标签关系表中增加若干条记录
        //2.1 获取专辑标签VO集合 转为专辑标签PO集合
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            //2.2 专辑标签关联专辑ID，保存
            albumAttributeValueVoList
                    .stream()
                    .forEach(albumAttributeValueVo -> {
                        //将VO转为PO
                        AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
                        albumAttributeValue.setAlbumId(albumId);
                        //保存专辑标签
                        albumAttributeValueMapper.insert(albumAttributeValue);
                    });
        }

        //3.固定为专辑保存四条统计信息
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_PLAY, 0);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_SUBSCRIBE, 0);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_BUY, 0);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_COMMENT, 0);

        //TODO 调用第三方“内容安全”服务接口进行内容审核，审核通过后发送Kafka消息通知搜索服务专辑上架
        if ("1".equals(albumInfo.getIsOpen())) { //模拟内容审核通过
            kafkaService.sendKafkaMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        }
    }

    /**
     * 保存专辑统计信息
     *
     * @param albumId  专辑ID
     * @param statType 统计类型 0401-播放量 0402-订阅量 0403-购买量 0403-评论数
     * @param num      数值
     */
    @Override
    public void saveAlbumStat(Long albumId, String statType, Integer num) {
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(num);
        albumStatMapper.insert(albumStat);
    }

    /**
     * 分页查询当前用户专辑列表
     *
     * @param pageInfo       分页对象
     * @param albumInfoQuery 查询条件
     * @return 分页对象
     */
    @Override
    public Page<AlbumListVo> getUserAlbumPage(Page<AlbumListVo> pageInfo, AlbumInfoQuery albumInfoQuery) {
        //1.获取专辑持久层对象，调用持久层对象方法（动态SQL+分页）
        return albumInfoMapper.getUserAlbumPage(pageInfo, albumInfoQuery);
    }

    @Autowired
    private TrackInfoMapper trackInfoMapper;

    /**
     * 删除专辑
     *
     * @param id
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void removeAlbumInfo(Long id) {
        //1.判断该专辑下是否包含声音如果包含则拒绝删除
        LambdaQueryWrapper<TrackInfo> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(TrackInfo::getAlbumId, id);
        Long count = trackInfoMapper.selectCount(queryWrapper);
        if (count > 0) {
            throw new GuiguException(400, "该专辑下仍然关联" + count + "个声音");
        }
        //2.根据主键删除专辑表（逻辑删除）
        albumInfoMapper.deleteById(id);

        //3.根据专辑ID删除专辑标签记录（逻辑删除）
        LambdaQueryWrapper<AlbumAttributeValue> attributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
        attributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        albumAttributeValueMapper.delete(attributeValueLambdaQueryWrapper);

        //4.根据专辑ID删除专辑统计记录（逻辑删除）
        LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId, id);
        albumStatMapper.delete(albumStatLambdaQueryWrapper);

        //5.发送Kafka消息通知搜索服务下架专辑

        kafkaService.sendKafkaMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());
    }

   @Autowired
    private RedisTemplate redisTemplate;

    /**//**
     * 根据专辑ID查询专辑信息（包含专辑标签列表）
     * 自定义分布式锁避免缓存击穿
     *
     * @param id
     * @return
     *//*
    @Override
    public AlbumInfo getAlbumInfo(Long id) {
        //1.优先从缓存Redis中获取业务数据
        //1.1 构建专辑信息业务数据key 形式：前缀+业务标识
        String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
        //1.2 从Redis中获取业务数据
        AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
        //1.3 命中缓存则执行返回，未命中缓存执行第二步
        if (albumInfo != null) {
            return albumInfo;
        }
        //2.尝试获取分布式锁
        //2.1 构建锁的Key 形式：前缀+业务标识+锁后缀
        String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
        //2.2 构建锁的Value 采用唯一ID避免不同线程互相释放锁
        String lockValue = IdUtil.randomUUID();
        //2.3 采用set ex nx命令尝试获取锁
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockValue, lockValue, 5, TimeUnit.SECONDS);
        if (flag) {
            //3.获取锁成功执行业务代码
            try {
                //3.1 从数据库中查询业务数据
                albumInfo = this.getAlbumInfoFromDB(id);
                //3.2 将查询业务数据放入缓存
                int randomTTL = RandomUtil.randomInt(200, 2000);
                redisTemplate.opsForValue().set(dataKey, albumInfo, RedisConstant.ALBUM_TIMEOUT + randomTTL, TimeUnit.SECONDS);
                //3.3 将结果返回
                return albumInfo;
            } finally {
                //4.释放锁
                String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                        "then\n" +
                        "    return redis.call(\"del\",KEYS[1])\n" +
                        "else\n" +
                        "    return 0\n" +
                        "end";
                DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
                redisScript.setScriptText(scriptText);
                redisScript.setResultType(Long.class);
                redisTemplate.execute(redisScript, Arrays.asList(lockKey), lockValue);
            }
        } else {
            //5.获取锁失败执行业务自旋
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return this.getAlbumInfo(id);
        }
    }*/

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 根据专辑ID查询专辑信息（包含专辑标签列表）
     * Redisson分布式锁避免缓存击穿
     *
     * @param id
     * @return
     */
    @Override
    public AlbumInfo getAlbumInfo(Long id) {
        //1.优先从缓存Redis中获取业务数据
        //1.1 构建专辑信息业务数据key 形式：前缀+业务标识
        String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
        //1.2 从Redis中获取业务数据
        AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
        //1.3 命中缓存则执行返回，未命中缓存执行第二步
        if (albumInfo != null) {
            return albumInfo;
        }
        //2.获取分布式锁,开启看门狗实现锁自动续期
        //2.1 构建锁的Key 形式：前缀+业务标识+锁后缀
        String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
        //2.2 创建锁对象
        RLock lock = redissonClient.getLock(lockKey);
        //2.3 获取锁 线程阻塞到获取锁成功为止
        lock.lock();
        //3. 执行业务代码
        try {
            //3.1 执行从DB中获取数据逻辑
            albumInfo = this.getAlbumInfoFromDB(id);
            //3.2 将查询结果放入缓存并且返回
            int randomTTL = RandomUtil.randomInt(200, 2000);
            redisTemplate.opsForValue().set(dataKey, albumInfo, RedisConstant.ALBUM_TIMEOUT + randomTTL, TimeUnit.SECONDS);
            return albumInfo;
        } finally {
            //3.3 业务执行完毕释放锁
            lock.unlock();
        }
    }

    /**
     * 根据专辑ID查询专辑信息（包含专辑标签列表）
     *
     * @param id
     * @return
     */
    @Override
    @GuiGuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX)
    public AlbumInfo getAlbumInfoFromDB(Long id) {
        //1.根据专辑主键ID查询专辑信息
        AlbumInfo albumInfo = albumInfoMapper.selectById(id);
        Assert.notNull(albumInfo, "专辑信息:{}不存在", id);

        //2.根据专辑ID查询专辑标签列表，将标签列表封装到专辑对象中
        LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(queryWrapper);
        if (CollectionUtil.isNotEmpty(albumAttributeValueList)) {
            albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
        }
        return albumInfo;
    }

    @Autowired
    private AlbumAttributeValueService albumAttributeValueService;

    /**
     * 修改专辑信息
     *
     * @param id
     * @param albumInfoVo
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateAlbumInfo(Long id, AlbumInfoVo albumInfoVo) {
        //1.修改专辑信息
        //1.1 将提交专辑信息VO转为PO对象-注意手动封装专辑ID
        AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);
        albumInfo.setId(id);
        //1.2 调用持久层修改
        albumInfoMapper.updateById(albumInfo);

        //2.修改专辑标签
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            //2.1 根据专辑ID先删除专辑标签关系记录
            LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
            albumAttributeValueMapper.delete(queryWrapper);

            //2.2 新增专辑标签关系记录
            List<AlbumAttributeValue> attributeValueList = albumAttributeValueVoList
                    .stream()
                    .map(albumAttributeValueVo -> {
                        AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
                        albumAttributeValue.setAlbumId(id);
                        return albumAttributeValue;
                    }).collect(Collectors.toList());
            //2.3 批量新增专辑标签关系
            albumAttributeValueService.saveBatch(attributeValueList);
        }

        //TODO 调用第三方“内容安全”服务接口进行内容审核，审核通过后发送Kafka消息通知搜索服务专辑上架
        if ("1".equals(albumInfo.getIsOpen())) { //模拟内容审核通过
            kafkaService.sendKafkaMessage(KafkaConstant.QUEUE_ALBUM_UPPER, id.toString());
        } else {
            kafkaService.sendKafkaMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());
        }
    }

    /**
     * 查询当前用户发布所有专辑列表
     *
     * @param userId
     * @return
     */
    @Override
    public List<AlbumInfo> getUserAllAlbumList(Long userId) {
        //1.构建查询条件对象
        LambdaQueryWrapper<AlbumInfo> queryWrapper = new LambdaQueryWrapper<>();
        //1.2.设置查询条件：用户ID
        queryWrapper.eq(AlbumInfo::getUserId, userId);
        //1.3.设置排序字段
        queryWrapper.orderByDesc(AlbumInfo::getId);
        //1.4.设置查询字段 设置select部分
        queryWrapper.select(AlbumInfo::getId, AlbumInfo::getAlbumTitle);
        //1.5.设置限制返回记录数 设置 limit 200
        queryWrapper.last("limit 200");
        //2.执行查询
        return albumInfoMapper.selectList(queryWrapper);
    }

    /**
     * 根据专辑ID查询专辑统计信息
     *
     * @param albumId
     * @return
     */
    @Override
    @GuiGuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX+":stat:")
    public AlbumStatVo getAlbumStatVo(Long albumId) {
        return albumInfoMapper.getAlbumStatVo(albumId);
    }

}
